#include "rtsp_connection.h"
#include "rtsp_context.h"
#include "utils/kmp.h"
#include "dispatcher/udp_dispatcher.h"
#include "dispatcher/tcp_dispatcher.h"

using namespace VLRTSP;




/**
 * @Author: CimZzz
 * @Date: 2022-01-11 15:42:12
 * 
 * 构造函数
 * @param connectionId 链接 Id
 * @param context RTSP 上下文
 * @param socket 套接字
 * @param closeCallback 链接关闭回调函数
 */
RTSPConnection::RTSPConnection(
	uint32_t connectionId,
	RTSPContext* context, 
	const std::shared_ptr<TCPSocket>& socket
): socket(socket), writeQueue([this] (std::shared_ptr<Buffer> data) {
	this->_onRecvWriteBuffer(data);
}), readThread([this]{
	this->_onReadRun();
}), dispatcher(nullptr), sessionId("")  {
	this->connectionId = connectionId;
	this->context = context;
	this->socketId = socket.get()->getSocketId();
	this->writeQueue.start();
	this->isClosed = false;
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-11 15:52:00
 * 
 * 析构函数
 */
RTSPConnection::~RTSPConnection() {
	this->close();
}


/**
 * @Author: CimZzz
 * @Date: 2022-01-13 19:23:31
 * 
 * 添加链接关闭回调函数
 * @param closeCallback 链接关闭回调
 * @return 如果添加回调成功返回 true
 */
bool RTSPConnection::addCloseCallback(CloseCallback closeCallback) {
	if(this->isClosed) {
		return false;
	}
	this->mutex.lock();
	if(this->isClosed) {
		this->mutex.unlock();
		return false;
	}
	this->closeCallbackVector.push_back(closeCallback);
	this->mutex.unlock();
	return true;
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-11 15:55:05
 * 
 * 关闭 RTSP 链接
 */
void RTSPConnection::close() {
	if(this->isClosed) {
		return;
	}
	this->mutex.lock();
	if(this->isClosed) {
		this->mutex.unlock();
		return;
	}
	this->isClosed = true;
	this->mutex.unlock();
	this->socket.get()->close();
	this->writeQueue.destroy();
	if(!this->closeCallbackVector.empty()) {
		for(auto iter = this->closeCallbackVector.begin(); iter != this->closeCallbackVector.end() ; iter ++) {
			(*iter)(this->connectionId, std::shared_ptr<std::string>(nullptr));
		}
		this->closeCallbackVector.clear();
	}
}


/**
 * @Author: CimZzz
 * @Date: 2022-01-13 11:31:52
 * 
 * 获取链接 Id
 */
uint32_t RTSPConnection::getConnectionId() {
	return this->connectionId;
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-11 16:25:10
 * 
 * 获取 Socket Id
 * @return socket id
 */
VL_SOCKET RTSPConnection::getSocketId() {
	return this->socketId;
}


/**
 * @Author: CimZzz
 * @Date: 2022-01-17 00:11:51
 * 
 * 获取写队列
 */
QueueThread<Buffer>* RTSPConnection::getWriteQueue() {
	return &this->writeQueue;
}

/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-01-14 11:21:35
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * 内部方法
 * 
 */


/**
 * @Author: CimZzz
 * @Date: 2022-01-12 09:24:37
 * 
 * 读取线程执行回调函数
 */
void RTSPConnection::_onReadRun() {
	TCPSocket* socketPtr = this->socket.get();
	// 字节缓冲区
	Buffer buffer;
	// 构造 KMP
	KMP kmp("\r\n\r\n");
	RTSPParser parser;
	uint32_t currentBufferIndex = 0;
	
	while(!isClosed) {
		uint8_t* bufferPtr = buffer.getBuffer() + currentBufferIndex;
		uint32_t bufferLength = buffer.getBufferLength() - currentBufferIndex;
		int64_t afterIndex = kmp.matchAfter(bufferPtr, bufferLength);
		if(afterIndex == -1) {
			if(buffer.getBufferLength() < 4) {
				currentBufferIndex = 0;
			} else {
				currentBufferIndex = buffer.getBufferLength() - 4;
			}
			if(socketPtr->read(&buffer) == -1) {
				VL_LOG("RTSPConnection", "读取 Socket 失效")
				this->close();
				break;
			}
			continue;
		} else {
			currentBufferIndex = 0;
		}
		std::shared_ptr<Buffer> respBuffer(new Buffer);
		// 解析 RTSP 数据
		if(!parser.parseHeader(&buffer)) {
			parser.buildServerErrorResponse(respBuffer.get());
			this->writeQueue.send(respBuffer);
			buffer.removeRange(0, afterIndex);
			continue;
		}

		VL_LOG("RTSPConnection", "收到 RTSP 请求: %s", parser.getMethodStr().c_str(), parser.getSuffix().c_str())
		if(parser.getMethod() == RTSPParser::RTCP) {
			// todo RTCP 处理
			buffer.removeRange(0, afterIndex);
			continue;
		}

		// 需要检查当前 Session 中是否存在 Suffix
		// 如果不存在应关闭当前链接

		std::string sessionId = parser.getSuffix();

		if(this->sessionId == "") {
			this->sessionId = sessionId;
		}
		else if(this->sessionId != sessionId) {
			VL_LOG("RTSPConnection", "Session Id 不匹配, current: %s, recv: %s", this->sessionId.c_str(), parser.getSuffix().c_str())
			parser.buildServerErrorResponse(respBuffer.get());
			this->writeQueue.send(respBuffer);
			buffer.removeRange(0, afterIndex);
			continue;
		}


		std::shared_ptr<MediaSession> session(this->context->getMediaSession(sessionId));
		if(session.get() == nullptr) {
			VL_LOG("RTSPConnection", "不存在对应的媒体会话: %s", parser.getSuffix().c_str())
			parser.buildNotFoundResponse(respBuffer.get());
			goto PARSER_END;
		}

		switch (parser.getMethod()) {
		case RTSPParser::OPTIONS: 
			this->_handleOPTIONS(parser, session, respBuffer);
			break;
		
		case RTSPParser::DESCRIBE: {
			this->_handleDESCRIBE(parser, session, respBuffer);
			break;
		}
		case RTSPParser::SETUP: {
			this->_handleSETUP(parser, session, respBuffer);
			break;
		}
		case RTSPParser::PLAY: {
			this->_handlePLAY(parser, session, respBuffer);
			break;
		}
		case RTSPParser::TEARDOWN: {
			this->_handleTEARDOWN(parser, session, respBuffer);
			// 写入完成后关闭链接
			this->writeQueue.sendRunnable([this] {
				this->close();
			});
			// 直接结束读取线程
			return;
		}
		case RTSPParser::GET_PARAMETER: {
			this->_handleGET_PARAMETER(parser, session, respBuffer);
			break;
		}
		default: {
			break;
		}
		}

		PARSER_END:
			this->writeQueue.send(respBuffer);
			buffer.removeRange(0, afterIndex);
	}
	
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-11 16:00:36
 * 
 * 消息队列接收到写数据
 * @param dataPtr 写数据指针
 */
void RTSPConnection::_onRecvWriteBuffer(std::shared_ptr<Buffer> buffer) {
	TCPSocket* socketPtr = this->socket.get();
	if(!socketPtr->write(buffer.get())) {
		// todo 写入失败，认作链接失效
		VL_LOG("RTSPConnection", "Socket 数据写入失败")
		this->close();
		return;
	};
}

/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-01-14 11:26:18
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * RTSP 指令处理方法
 * 
 */


/**
 * @Author: CimZzz
 * @Date: 2022-01-14 11:23:27
 * 
 * 处理 OPTIONS 指令
 * @param parser RTSP 解析器
 * @param mediaSession 媒体会话
 * @param respBuffer 响应字节缓冲区
 */
void RTSPConnection::_handleOPTIONS(RTSPParser& parser, std::shared_ptr<MediaSession>& mediaSession, std::shared_ptr<Buffer>& respBuffer) {
	parser.buildOptionsResponse(respBuffer.get());
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-14 11:25:15
 * 
 * 处理 DESCRIBE 指令
 * @param parser RTSP 解析器
 * @param mediaSession 媒体会话
 * @param respBuffer 响应字节缓冲区
 */
void RTSPConnection::_handleDESCRIBE(RTSPParser& parser, std::shared_ptr<MediaSession>& mediaSession, std::shared_ptr<Buffer>& respBuffer) {
	std::shared_ptr<Buffer> sdpBuffer(new Buffer);
	mediaSession.get()->buildSDP(parser.getIp(), sdpBuffer.get());
	parser.buildDescribeResponse(respBuffer.get(), sdpBuffer.get());
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-14 11:25:28
 * 
 * 处理 SETUP 指令
 * @param parser RTSP 解析器
 * @param mediaSession 媒体会话
 * @param respBuffer 响应字节缓冲区
 */
void RTSPConnection::_handleSETUP(RTSPParser& parser, std::shared_ptr<MediaSession>& mediaSession, std::shared_ptr<Buffer>& respBuffer) {
	// Find MediaSource
	std::shared_ptr<MediaSource> mediaSouce(mediaSession.get()->findMediaSource(parser.getTrackId()));
	if(mediaSouce.get() == nullptr) {
		// 没有找到对应资源
		VL_LOG("RTSPConnection", "没有找到 Media 资源: trackId = %d", parser.getTrackId())
		parser.buildNotFoundResponse(respBuffer.get());
		return;
	}

	switch(parser.getTransportMode()) {
		case Transport_UDP: {
			VL_LOG("RTSPConnection", "不支持的传输类型: UDP")
			parser.buildUnsupportedTransportResponse(respBuffer.get());
			break;
			this->dispatcher.reset(new UDPDispatcher(
				parser.getIp(),
				this->context->generateDispatcherId(),
				parser.getRTPSendPoint(), 
				parser.getRTCPSendPoint(),
				mediaSouce)
			);
			UDPDispatcher* dispatcher = (UDPDispatcher*)this->dispatcher.get();
			if(!dispatcher->init()) {
				// 分发器初始化失败
				VL_LOG("RTSPConnection", "UDP 分发器初始化失败")
				parser.buildServerErrorResponse(respBuffer.get());
				break;
			}
			VL_LOG("RTSPConnection", "UDP 分发器初始化成功")
			parser.buildSetupByUdpResponse(respBuffer.get(), dispatcher->getServerRTPPort(), dispatcher->getServerRTCPPort(), dispatcher->getDispatcherId());
			break;
		}
		case Transport_TCP: {
			this->dispatcher.reset(new TCPDispatcher(
				&this->writeQueue,
				this->context->generateDispatcherId(),
				parser.getRTPSendPoint(), 
				parser.getRTCPSendPoint(),
				mediaSouce)
			);
			MediaDispatcher* dispatcher = this->dispatcher.get();
			if(!dispatcher->init()) {
				// 分发器初始化失败
				VL_LOG("RTSPConnection", "TCP 分发器初始化失败")
				parser.buildServerErrorResponse(respBuffer.get());
				break;
			}
			VL_LOG("RTSPConnection", "TCP 分发器初始化成功")
			parser.buildSetupByTcpResponse(respBuffer.get(), dispatcher->getDispatcherId());
			break;
		}
		case Transport_MULTICAST: {
			// 目前不支持多播
			VL_LOG("RTSPConnection", "不支持的传输类型: 组播")
			parser.buildUnsupportedTransportResponse(respBuffer.get());
			break;
		}
		case Transport_NONE: {
			VL_LOG("RTSPConnection", "不支持的传输类型: Unknown")
			parser.buildUnsupportedTransportResponse(respBuffer.get());
			break;
		}
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-14 11:25:37
 * 
 * 处理 PLAY 指令
 * @param parser RTSP 解析器
 * @param mediaSession 媒体会话
 * @param respBuffer 响应字节缓冲区
 */
void RTSPConnection::_handlePLAY(RTSPParser& parser, std::shared_ptr<MediaSession>& mediaSession, std::shared_ptr<Buffer>& respBuffer) {
	MediaDispatcher* dispatcher = this->dispatcher.get();
	if(dispatcher == nullptr || dispatcher->getDispatcherId() != parser.getDispatcherId()) {
		parser.buildServerErrorResponse(respBuffer.get());
		return;
	}
	dispatcher->start();
	parser.buildPlayResponse(respBuffer.get());
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-14 11:25:44
 * 
 * 处理 TEARDOWN 指令
 * @param parser RTSP 解析器
 * @param mediaSession 媒体会话
 * @param respBuffer 响应字节缓冲区
 */
void RTSPConnection::_handleTEARDOWN(RTSPParser& parser, std::shared_ptr<MediaSession>& mediaSession, std::shared_ptr<Buffer>& respBuffer) {
	parser.buildTeardownResponse(respBuffer.get());
}

/**
 * @Author: CimZzz
 * @Date: 2022-01-14 11:25:55
 * 
 * 处理 GET_PARAMETER 指令
 * @param parser RTSP 解析器
 * @param mediaSession 媒体会话
 * @param respBuffer 响应字节缓冲区
 */
void RTSPConnection::_handleGET_PARAMETER(RTSPParser& parser, std::shared_ptr<MediaSession>& mediaSession, std::shared_ptr<Buffer>& respBuffer) {
	
}